package com.gitee.yanfanvip.cluster;

import java.io.DataInput;
import java.io.DataOutput;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.jgroups.Address;
import org.jgroups.Global;
import org.jgroups.Header;
import org.jgroups.Message;
import org.jgroups.annotations.MBean;
import org.jgroups.blocks.MethodCall;
import org.jgroups.conf.ClassConfigurator;
import org.jgroups.stack.IpAddress;
import org.jgroups.stack.Protocol;
import org.jgroups.util.MessageBatch;
import org.rocksdb.RocksDBException;

import com.gitee.yanfanvip.exception.RemoteException;
import com.gitee.yanfanvip.interfaces.DatabaseWrite;
import com.gitee.yanfanvip.model.Row;

@MBean(description="远程集群调用")
public class RemoteCluster extends Protocol implements DatabaseWrite{
	protected static final short Protocol_ID = 11502;
	protected static final short Header_ID = 21502;
	static{
		ClassConfigurator.add(Header_ID, Remote_Header.class);
		ClassConfigurator.addProtocol(Protocol_ID, RemoteCluster.class);
	}
	protected final HashMap<String, CompletableFuture<Object>> call = new HashMap<String, CompletableFuture<Object>>();
	protected final HashMap<String,RemoteSender> senders = new HashMap<String, RemoteSender>();
	protected String cluster;
	protected ClusterDatabase clusterDatabase;
	
	public RemoteCluster(String cluster) {
		this.cluster = cluster;
	}

	/**
	 * 添加远端站点
	 * @throws Exception 
	 */
	public void addSite(String cluster, String remoteServer) throws Exception {
		if(cluster.equals(this.cluster)){ return; }
		String[] services = remoteServer.split(",");
		List<IpAddress> addresses = new ArrayList<IpAddress>();
		for (String service : services) {
			addresses.add(new IpAddress(service));
		}
		senders.put(cluster.toUpperCase(), new RemoteSender(cluster, addresses));
	}

	@Override
	public void start() throws Exception {
		clusterDatabase = getProtocolStack().findProtocol(ClusterDatabase.class);
	}
	
	/**
	 * 接收到
	 */
	@Override
	public Object up(Message message) {
		Remote_Header header = message.getHeader(Header_ID);
		if(header != null){
			if(header.call) {
				Message report = new Message(message.getSrc(), "").putHeader(Header_ID, new Remote_Header(header.id, false, true));
				getTransport().down(report);
				Object result;
				try {
					result = call(header, message);
				} catch (Exception e) {
					result = e;
				}
				Message msg = new Message(message.getSrc(), result);
				msg.putHeader(Header_ID, new Remote_Header(header.id, false, false));
				getTransport().down(msg);
			}else if(header.report){
				back(header.id + ":report", message);
			}else {
				back(header.id + ":call", message);
			}
			return null;
		}
		return super.up(message);
	}
	
	private Object call(Remote_Header header, Message message) throws Exception {
		Object msg = message.getObject();
		if(msg instanceof MethodCall) {
			MethodCall call = (MethodCall) msg;
			return call.invoke(new CallMethods());
		}else {
			return new IllegalAccessError("bad request");
		}
	}
	
	private void back(String key, Message message) {
		CompletableFuture<Object> future = call.get(key);
		if(future == null) { return; }
		Object obj = message.getObject();
		if(obj != null && obj instanceof Throwable) {
			future.completeExceptionally((Throwable)obj);
		}else {
			future.complete(message.getObject());
		}
	}

	@Override
	public void up(MessageBatch batch) {
		for (Message message : batch) {
			Remote_Header header = message.getHeader(Header_ID);
			if(header != null){
				batch.remove(message);
				up(message);
			}
		}
		if(!batch.isEmpty()){ up_prot.up(batch); }
	}
	
	private Object remote(Address address, MethodCall method, long timeout) throws Exception {
		Remote_Header header = new Remote_Header(true, false);
		try {
			CompletableFuture<Object> reportfuture = new CompletableFuture<Object>();
			call.put(header.id + ":report", reportfuture);
			CompletableFuture<Object> future = new CompletableFuture<Object>();
			call.put(header.id+ ":call", future);
			getTransport().down(new Message(address, method).putHeader(Header_ID, header));
			reportfuture.get(1000, TimeUnit.MILLISECONDS);
			return future.get(timeout, TimeUnit.MILLISECONDS);
		}finally {
			call.remove(header.id+ ":call");
			call.remove(header.id+ ":report");
		}
	}
	
	public class CallMethods{

		public boolean put(String keyspaces, String key, byte[] value) throws RocksDBException {
			clusterDatabase._cluster_put(keyspaces, key, value);
			return true;
		}

		public boolean putAll(String keyspaces, Map<String, byte[]> datas) throws RocksDBException {
			clusterDatabase._cluster_putAll(keyspaces, datas);
			return true;
		}
		
		public boolean putAll(String keyspaces, Set<Row> rows) throws RocksDBException {
			Map<String, byte[]> datas = new HashMap<>();
			for (Row row : rows) {
				datas.put(row.getKey(), row.getValue());
			}
			clusterDatabase._cluster_putAll(keyspaces, datas);
			return true;
		}

		public boolean remove(String keyspaces, String key) throws RocksDBException {
			clusterDatabase._cluster_remove(keyspaces, key);
			return true;
		}

		public boolean clear(String keyspaces) throws RocksDBException {
			clusterDatabase._cluster_clear(keyspaces);
			return true;
		}
	}

	public class RemoteSender implements DatabaseWrite{
		String cluster;
		List<IpAddress> addresss;
		boolean lived;//是否存活
		IpAddress liveNode;
		long lastUpdated;//上次通信时间
		
		public RemoteSender(String cluster, List<IpAddress> addresss) {
			this.cluster = cluster;
			this.addresss = addresss;
		}
		
		private boolean call(MethodCall method) {
			IpAddress liveNode = null;
			boolean flag = false;
			Throwable t = null;
			for (IpAddress address : addresss) {
				try {
					flag = (boolean) remote(address, method, 60 * 1000);
					liveNode = address;
					break;
				} catch (Exception e) { t = e; } 
			}
			if(liveNode == null){
				this.liveNode = liveNode;
				lived = false;
				throw new RemoteException(this.cluster, t);
			}
			if(!lived){
				//网络恢复之后, 同步数据
				new Thread(new Runnable() { 
					public void run() { 
						try { fullData(null); } catch (RocksDBException e) { } 
					} 
				}).start();
			}
			addresss.set(0, liveNode);
			this.liveNode = liveNode;
			lived = true;
			lastUpdated = System.currentTimeMillis();
			return flag;
		}
		
		private Method getMethod(String name, Class<?>... parameterTypes){
			try {
				return CallMethods.class.getMethod(name, parameterTypes);
			} catch (NoSuchMethodException | SecurityException e) {
				throw new RemoteException(this.cluster, e);
			}
		}
		
		public void fullData(String keyspaces) throws RocksDBException{
			if(keyspaces == null){
				Set<String> keyspacess = clusterDatabase.keyspaces();
				for (String keyspace : keyspacess) { this.fullData(keyspace); }
			}else{
				Method method = getMethod("putAll", String.class, Set.class);
				clusterDatabase.forEach(keyspaces, 10000, rows->{ call(new MethodCall(method, keyspaces, rows)); });
			}
		}

		@Override
		public void put(String keyspaces, String key, byte[] value) throws RocksDBException {
			call(new MethodCall(getMethod("put", String.class, String.class, byte[].class), keyspaces, key, value));
		}

		@Override
		public void putAll(String keyspaces, Map<String, byte[]> datas) throws RocksDBException {
			call(new MethodCall(getMethod("putAll", String.class, Map.class), keyspaces, datas));
		}

		@Override
		public void remove(String keyspaces, String key) throws RocksDBException {
			call(new MethodCall(getMethod("remove", String.class, String.class), keyspaces, key));
		}

		@Override
		public void clear(String keyspaces) throws RocksDBException {
			call(new MethodCall(getMethod("clear", String.class), keyspaces));
		}
	}
	
	public static class Remote_Header extends Header{
	    public Supplier<? extends Header> create() { return Remote_Header::new; }
	    private Random random = new Random();
	    private long id;
	    private boolean call;
	    private boolean report;
	    public Remote_Header() { this.id = random.nextLong(); }
	    public Remote_Header(boolean call, boolean report) { this.id = random.nextLong(); this.call = call; this.report = report; }
	    public Remote_Header(long id, boolean call, boolean report) { this.id = id; this.call = call; this.report = report; }
	    public short getMagicId() {return Header_ID;}
	    public int serializedSize() { return Global.INT_SIZE; }
	    public boolean call() { return call; }
	    public boolean report() { return report; }
		public void writeTo(DataOutput out) throws Exception { 
			out.writeLong(id);
			out.writeBoolean(call);
			out.writeBoolean(report);
		}
		public void readFrom(DataInput in) throws Exception {
			this.id = in.readLong();
			this.call = in.readBoolean();
			this.report = in.readBoolean();
		}
	}
	
	
	//================================================
	@Override
	public void put(String keyspaces, String key, byte[] value) throws RocksDBException {
		for (RemoteSender sender : senders.values()) {
			sender.put(keyspaces, key, value);
		}
	}

	@Override
	public void putAll(String keyspaces, Map<String, byte[]> datas) throws RocksDBException {
		for (RemoteSender sender : senders.values()) {
			sender.putAll(keyspaces, datas);
		}
	}

	@Override
	public void remove(String keyspaces, String key) throws RocksDBException {
		for (RemoteSender sender : senders.values()) {
			sender.remove(keyspaces, key);
		}
	}

	@Override
	public void clear(String keyspaces) throws RocksDBException {
		for (RemoteSender sender : senders.values()) {
			sender.clear(keyspaces);
		}
	}
	
}
